package com.jeffse.light.stream.core.function.base;

import com.jeffse.light.stream.core.data.context.*;
import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.pojo.OutTag;

import java.util.UUID;

/**
 * keyedProcessFunction 在远程提交时不能使用匿名类提交，必须用新的类 继承此类实现，否则序列化的时候会产生错误
 * @param <KEY>
 */
public abstract class KeyedProcessFunction<KEY, T, O>  extends TimerContext<KEY, T, O> implements AbstractProcessFunction<T, O> {
    protected RuntimeContext runtimeContext;
    protected Context ctx;
    protected String uuid;
    protected OutTag outTag;
    protected OperatorDescriptor nextOperatorDescriptor = null;

    @Override
    public void setNextOperatorDescriptor(OperatorDescriptor nextOperatorDescriptor) {
        this.nextOperatorDescriptor = nextOperatorDescriptor;
    }

    @Override
    public OperatorDescriptor getNextOperatorDescriptor() {
        return nextOperatorDescriptor;
    }

    @Override
    public void setCtx(Context ctx) {
        this.ctx = ctx;
    }

    @Override
    public Context getCtx() {
        return ctx;
    }

    @Override
    public String getUUID() {
        return uuid;
    }

    @Override
    public void setOutTag(OutTag outTag) {
        this.outTag = outTag;
    }

    @Override
    public OutTag getOutTag() {
        return outTag;
    }

    @Override
    public RuntimeContext getRuntimeContext() {
        return runtimeContext;
    }

    @Override
    public void setRuntimeContext(RuntimeContext runtimeContext) {
        this.runtimeContext = runtimeContext;
    }

    public KeyedProcessFunction(){
        open();
    }

    public void open(){
        runtimeContext = new StreamingRuntimeContext();
        setProcessContext(this);
        uuid = UUID.randomUUID().toString();
    }

    @Override
    public void collect(O element, OutTag outTag) throws Exception {
        ctx.collect(element, outTag);
    }

    @Override
    public void collect(O element) throws Exception {
        ctx.collect(element, nextOperatorDescriptor, outTag);
    }
}